#include "my_disk_index.h"
#include "comm/my_mem.h"

// digest64
#define MY_DISK_INDEX_START_IDX(index, digest)              \
    ((digest % index->slot_num) &~ (MY_DISK_ONE_BUCKET_MASK))

#define MY_DISK_SLOT_AT_ITEM(index, slot_idx)              \
    ((my_disk_index_item_t *)(index->table) + slot_idx)

static void
my_disk_index_insert_helper(my_disk_index_t *index, int64_t start_idx,
    int64_t bucket_idx, uint32_t digest32, my_disk_index_item_t *item) {

    my_disk_index_item_t *it;
    my_bool_t found = MY_FALSE;
    for (int i = 0; i < MY_DISK_ONE_BUCKET_ITEMS; i++) {
        it = MY_DISK_SLOT_AT_ITEM(index, start_idx + i);
        if (it->digest32 == digest32) {
            found = MY_TRUE;
            break;
        }
    }

    if (found) {
        *it = *item;
        return;
    }

    my_bool_t found_empty_slot = MY_FALSE;
    for (int i = 0; i < MY_DISK_ONE_BUCKET_ITEMS; i++) {
        it = MY_DISK_SLOT_AT_ITEM(index, start_idx + i);
        if (!my_disk_vinfo_vaild_index_item(index->vinfo, it)) {
            found_empty_slot = MY_TRUE;
            break;
        }
    }

    if (!found_empty_slot) {
        int64_t evict = *(index->bucket_evict_slots+bucket_idx);
        *(index->bucket_evict_slots+bucket_idx) = 
            (*(index->bucket_evict_slots+bucket_idx) + 1) % MY_DISK_ONE_BUCKET_ITEMS;
        it = MY_DISK_SLOT_AT_ITEM(index, start_idx+evict);
    }

    *it = *item;
}

void
my_disk_index_insert(my_disk_index_t *index, my_cache_key_t *key,
    my_disk_index_item_t *item) {

    int64_t start_idx = MY_DISK_INDEX_START_IDX(index, key->digest);
    int64_t bucket_idx = start_idx / MY_DISK_ONE_BUCKET_ITEMS;
    int64_t group_idx = bucket_idx / MY_DISK_ONE_GROUP_BUCKETS;

    my_rwlock_wlock(index->group_locks + group_idx);
    my_disk_index_insert_helper(index, start_idx, bucket_idx, key->digest32, item);
    *(index->group_dirtys+group_idx) = *(index->group_dirtys+group_idx) + 1;
    my_rwlock_wunlock(index->group_locks + group_idx);
}

static my_bool_t
my_disk_index_lookup_helper(my_disk_index_t *index, int64_t start_idx,
    uint32_t digest32, my_disk_index_item_t *item) {

    my_disk_index_item_t *it;
    for (int i = 0; i < MY_DISK_ONE_BUCKET_ITEMS; i++) {
        it = MY_DISK_SLOT_AT_ITEM(index, start_idx+i);
        if (it->digest32 == digest32) {
            if (my_disk_vinfo_vaild_index_item(index->vinfo, it)) {
                *item = *it;
                return MY_TRUE;
            }
            return MY_FALSE;
        }
    }
    return MY_FALSE;
}

my_bool_t
my_disk_index_lookup(my_disk_index_t *index, my_cache_key_t *key,
    my_disk_index_item_t *item) {

    int64_t start_idx = MY_DISK_INDEX_START_IDX(index, key->digest);
    int64_t bucket_idx = start_idx / MY_DISK_ONE_BUCKET_ITEMS;
    int64_t group_idx = bucket_idx / MY_DISK_ONE_GROUP_BUCKETS;

    my_bool_t res;
    my_rwlock_rlock(index->group_locks + group_idx);
    res = my_disk_index_lookup_helper(index, start_idx, key->digest32, item);
    my_rwlock_runlock(index->group_locks + group_idx);
    return res;
}

static my_bool_t
my_disk_index_remove_helper(my_disk_index_t *index, int64_t start_idx,
    uint32_t digest32) {

    my_disk_index_item_t *it;
    for (int i = 0; i < MY_DISK_ONE_BUCKET_ITEMS; i++) {
        it = MY_DISK_SLOT_AT_ITEM(index, start_idx + i);
        if (it->digest32 == digest32) {
            it->digest32 = 0;   // clear
            return MY_TRUE;
        }
    }

    return MY_FALSE;
}

my_bool_t
my_disk_index_remove(my_disk_index_t *index, my_cache_key_t *key) {
    int64_t start_idx = MY_DISK_INDEX_START_IDX(index, key->digest);
    int64_t bucket_idx = start_idx / MY_DISK_ONE_BUCKET_ITEMS;
    int64_t group_idx = bucket_idx / MY_DISK_ONE_GROUP_BUCKETS;

    my_bool_t res;
    my_rwlock_wlock(index->group_locks + group_idx);
    res = my_disk_index_remove_helper(index, start_idx, key->digest32);
    if (res) {
        *(index->group_dirtys+group_idx) = *(index->group_dirtys+group_idx) + 1;
    }
    my_rwlock_wunlock(index->group_locks + group_idx);
    return res;
}

my_bool_t
my_disk_index_open(my_disk_index_t *index, my_disk_vinfo_t *vinfo,
    char *table, int64_t table_size) {

    if (table_size % MY_DISK_ONE_GROUP_SIZE) {
        MY_LOG_FATAL("disk_index", "invalid table size: must to multiplier of "
        "MY_DISK_ONE_GROUP_SIZE, table_size: %d", table_size);
    }

    int64_t slot_num = table_size / MY_DISK_INDEX_ITEM_SIZE;
    int64_t bucket_num = slot_num / MY_DISK_ONE_BUCKET_ITEMS;
    int64_t group_num = bucket_num / MY_DISK_ONE_GROUP_BUCKETS;

    my_rwlock_t *group_locks = my_malloc(sizeof(my_rwlock_t) * group_num);
    for (int i = 0; i < group_num; i++) {
        my_rwlock_init(group_locks + i);
    }

    int64_t *group_dirtys = my_malloc(sizeof(int64_t) * group_num);
    bzero(group_dirtys, sizeof(int64_t) * group_num);

    int64_t *bucket_evict_slots = my_malloc(sizeof(int64_t) * bucket_num);
    bzero(bucket_evict_slots, sizeof(int64_t) * bucket_num);

    char *flush_buf = my_malloc(MY_DISK_ONE_GROUP_SIZE);

    index->vinfo = vinfo;
    index->slot_num = slot_num;
    index->group_num = group_num;
    index->group_locks = group_locks;
    index->group_dirtys = group_dirtys;
    index->bucket_num = bucket_num;
    index->bucket_evict_slots = bucket_evict_slots;
    index->table = table;
    index->table_size = table_size;
    index->flush_buf = flush_buf;

    // TODO: start timer task for flush index table

    index->open = MY_TRUE;
    return MY_TRUE;
}

void
my_disk_index_close(my_disk_index_t *index) {
    if (!index->open) {
        return;
    }

    // TODO: stop timer task for flush index table

    if (index->flush_buf) {
        my_free(index->flush_buf);
    }

    if (index->bucket_evict_slots) {
        my_free(index->bucket_evict_slots);
    }

    if (index->group_dirtys) {
        my_free(index->group_dirtys);
    }

    if (index->group_locks) {
        my_free(index->group_locks);
    }

    index->open = MY_FALSE;
}
